package com.zdp.demo.mq;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import java.util.Collection;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * <h1>RocketMQ发送消息工具类</h1>
 */
@RequiredArgsConstructor
@Slf4j
@Component
public class MessageProducer {

    private final RocketMQTemplate rocketMQTemplate;

    /**
     * <h2>同步发送消息</h2>
     * @param topic 主题
     * @param tag 标签
     * @param withPayload 消息内容
     * @return msgId
     */
    public <T> String syncSendMessage(String topic, String tag, T withPayload) {
        Assert.isTrue(StrUtil.isNotBlank(topic) && ObjectUtil.isNotNull(withPayload),"withPayload, or topic is null");
        Message<T> build = MessageBuilder.withPayload(withPayload).build();
        String dest = String.format("%s:%s",topic,tag);
        SendResult sendResult = rocketMQTemplate.syncSend(dest, build);
        log.info("发送消息：{} 发送结果：{}", withPayload, sendResult);
        SendStatus sendStatus =  sendResult.getSendStatus();
        Assert.isTrue(SendStatus.SEND_OK.equals(sendStatus), "MQ消息发送失败");
        return sendResult.getMsgId();
    }

    /**
     * <h2>同步发送消息</h2>
     * @param topic 主题
     * @param tag 标签 用于服务端消息过滤
     * @param key msgKey 用于消息在业务层面的唯一标识
     * @param withPayload 消息内容
     * @return msgId
     */
    public <T> String syncSendMQMessage(String topic, String tag, String key, T withPayload) {
        Message<T> build = MessageBuilder.withPayload(withPayload).setHeader(RocketMQHeaders.KEYS,key).build();
        String dest = String.format("%s:%s",topic,tag);
        SendResult sendResult = rocketMQTemplate.syncSend(dest, build);
        log.info("发送消息：{} 发送结果：{}", withPayload, sendResult);
        SendStatus sendStatus =  sendResult.getSendStatus();
        if (!Objects.equals(sendStatus,SendStatus.SEND_OK)) {
            throw new RuntimeException("发送mq异常");
        }
        return sendResult.getMsgId();
    }

    /**
     * <h2>发送批量消息</h2>
     * @param topic 主题
     * @param tag 标签
     * @param withPayloads 消息内容
     * @return msgId
     */
    public <T> String syncSendBatchMessage(String topic, String tag, Collection<T> withPayloads) {
        Assert.isTrue(StrUtil.isNotBlank(topic) && CollUtil.isNotEmpty(withPayloads),"withPayloads, or topic is null");
        Collection<Message<T>> messages = withPayloads.stream().map(item -> MessageBuilder.withPayload(item).build()).collect(Collectors.toList());
        String dest = String.format("%s:%s",topic,tag);
        SendResult sendResult = rocketMQTemplate.syncSend(dest, messages);
        log.info("发送消息：{} 发送结果：{}", withPayloads, sendResult);
        SendStatus sendStatus =  sendResult.getSendStatus();
        Assert.isTrue(SendStatus.SEND_OK.equals(sendStatus), "MQ消息发送失败");
        return sendResult.getMsgId();
    }
}
